package cn.doitedu.utils;

import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import redis.clients.jedis.Jedis;

import java.io.IOException;
import java.sql.DriverManager;

/**
 * @create: 2021-11-11 20:19
 * @author: 静如海的男人
 * @program: eagle-markting
 **/

public class ConnectionUtils {
    //Hbase的连接
    private static Connection conn1;
    //ClickHouse的连接
    private static java.sql.Connection conn2;
    //DorisDB的连接
    private static java.sql.Connection conn3;


    //获取Hbase的连接
    public static Connection createHbase(ParameterTool parameterTool) throws IOException {
        Configuration conf = HBaseConfiguration.create();

        conf.set("hbase.zookeeper.quorum", parameterTool.getRequired("hbase.zookeeper.quorum"));
        conn1 = ConnectionFactory.createConnection(conf);
        return conn1;

    }

    //获取Clickhouse的连接
    public static java.sql.Connection createClickHouse(ParameterTool parameterTool) throws Exception {
        Class.forName("ru.yandex.clickhouse.ClickHouseDriver");// 注册驱动
        conn2 = DriverManager.getConnection(parameterTool.getRequired("clickhouse.url"));//获取连接
        return conn2;
    }

    //获取DorisDB的连接
    public static java.sql.Connection createDorisDB(ParameterTool parameterTool) throws Exception {
        //url
        String url = parameterTool.getRequired("DorisDB.url");
        //用户
        String user = parameterTool.get("DorisDB.user", "root");
        //密码
        String password = parameterTool.get("DorisDB.password", "");

        Class.forName("com.mysql.jdbc.Driver");
        conn3 = DriverManager.getConnection(url, user, password);
        return conn3;
    }

    //获取Redis的连接
    public static Jedis getJedis(ParameterTool parameterTool){
        //主机名
        String host = parameterTool.getRequired("redis.host");
        //端口号
        int port = parameterTool.getInt("redis.port", 6379);
        //密码
        String password = parameterTool.getRequired("redis.password");
        //数据库
        int db = parameterTool.getInt("redis.db",4);

        //创建jedis实例
        Jedis jedis = new Jedis(host,port);
        //设置密码
        jedis.auth(password);
        //选择数据库
        jedis.select(db);
        return jedis;
    }

    //关闭所有连接
    public static void close() throws Exception {
        if (conn1!=null){
            conn1.close();
        }
        if (conn2!=null){
            conn2.close();
        }
        if (conn3!=null){
            conn3.close();
        }
    }
}
